package com.navinfo.tripanalysis;

import com.navinfo.tripanalysis.common.util.DateUtils;
import com.navinfo.tripanalysis.pojo.*;
import com.navinfo.tripanalysis.service.*;
import com.navinfo.tripanalysis.service.impl.ConfigLoadServiceYaml;
import com.navinfo.tripanalysis.util.CommonUtils;
import org.apache.commons.cli.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.*;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.*;
import java.net.URISyntaxException;
import java.util.*;

import static com.navinfo.tripanalysis.util.SparkAppUtils.getSparkBuilder;

public class TripAbnormalStatictisApplication {

    private static final Logger logger = LoggerFactory.getLogger(TripAbnormalStatictisApplication.class);

    /**
     * 开始时间毫秒数
     */
    static Long start;
    /**
     * 结束时间毫秒数
     * */
    static Long end;
    /**
     * 一天的毫秒数
     */
    static final Long DAY_MS = 86400000L;

    public static void main(String[] args) throws Exception {

        logger.error("开始统计速度传感器异常行程，args:{}", Arrays.toString(args));

        //加载配置
        Options options = setCmdOptions();
        CommandLine cmd = parseCmdOptions(options, args);
        Properties config = loadProps(cmd);
        parseDays(args, cmd);

        //初始化SparkSession
        SparkSession spark = getSparkBuilder(config).getOrCreate();
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());

        //创建相关服务
        ServiceFactory factory = new ServiceFactory(config);
        LoadAbnormalProtocolPointDataService loadProtocolPoint = factory.createLoadAbnormalProtocolPointService();
        LoadAbnormalTripStatictisDataService insertyTripData = factory.createLoadAbnormalTripStatictisService();
        //DataBatchService dataBatchService = factory.createDataBatchService();

        //创建参数列表
        LoadDataParam param = new LoadDataParam();

        //按天读取公共协议数据
        for(long sTime=start; sTime<end; sTime+= DAY_MS){

            //设置日期参数
            final Date currentDay = new Date(sTime);
            param.setDay(currentDay);

            //最终的输出结果
            List<JavaRDD<AbnormalProtocolPoint>> outDataRDDs = new ArrayList();

            //按批次处理数据
            //List<LoadDataParam> batchParams = dataBatchService.createBatch(param);

            //for (LoadDataParam oneBatchParam : batchParams) {

                //logger.error("========>oneBatchParam:{}", oneBatchParam);

            //加载公共协议数据开始
            //JavaPairRDD<行程ID,位置点>
            JavaPairRDD<Tuple2<Long,Long>, AbnormalProtocolPoint> ppointRDD = loadProtocolPoint.load(spark,jsc,param);
            //按行程ID汇总
            JavaPairRDD<Tuple2<Long,Long>,Iterable<AbnormalProtocolPoint>> ppointByTridRDD = ppointRDD.groupByKey();
            //判断是否有符合传感器异常条件行程ID
            JavaPairRDD<Long,Long> tripIdsRDD = getTripIds(ppointByTridRDD);

            /*
            //创建异常行程ID广播变量
            final Broadcast<Map<Integer,Integer>> AbnormalTripIdsBrocast = jsc.broadcast(tripIdsRDD.collectAsMap());
            //加载行程数据
            */
            //注册临时表
            //String tempTableName = "AbnormalTripids";
            String tempTableName = config.getProperty("load.data.trip.hive.Temptablename");
            RegisterTempTable(tripIdsRDD,tempTableName,spark);

            //插入行程数据
            insertyTripData.insert(spark,jsc,param);
        }


    }

    /**
     * 生成异常行程ID和终端ID的临时表
     * @param tripIdsRDD
     * @param tempTableName
     * @param spark
     */
    private static void RegisterTempTable(JavaPairRDD<Long, Long> tripIdsRDD, String tempTableName ,SparkSession spark) {

        //创建RDD<Row>
        JavaRDD<Row> tableRDD = tripIdsRDD.map(tids -> RowFactory.create(tids._1,tids._2));
        //创建Strunct
        List<StructField> abnormalIds = new ArrayList<>();
        abnormalIds.add(DataTypes.createStructField("tid",DataTypes.LongType,true));
        abnormalIds.add(DataTypes.createStructField("tripid",DataTypes.LongType,true));
        StructType structtype = DataTypes.createStructType(abnormalIds);

        Dataset<Row> abnorIds =  spark.createDataFrame(tableRDD,structtype);
        //注册临时表
        abnorIds.createOrReplaceTempView(tempTableName);
    }


    /**
     * 过滤出符合速度传感器异常行程的终端id和行程id
     * @param ppointByTripidRDD
     * @return
     */
    private static JavaPairRDD<Long,Long> getTripIds(JavaPairRDD<Tuple2<Long,Long>, Iterable<AbnormalProtocolPoint>> ppointByTripidRDD){

        //过滤非速度传感器异常行程
        JavaPairRDD<Tuple2<Long,Long>, Iterable<AbnormalProtocolPoint>> filterTripidsRDD =
                //ppointByTripidRDD.filter(new Function<Tuple2<Tuple2<Long,Long>, Iterable<AbnormalProtocolPoint>>, Boolean>() {
                ppointByTripidRDD.filter(new Function<Tuple2<Tuple2<Long, Long>, Iterable<AbnormalProtocolPoint>>, Boolean>() {
                    private static final long serialVersionUID = 1L;
                        @Override
                        public Boolean call(Tuple2<Tuple2<Long, Long>, Iterable<AbnormalProtocolPoint>> tuple) throws Exception {

                            //long tripId = tuple._1._1;
                            Iterator points = tuple._2.iterator();
                            int abnormalPoints = 0;
                            while (points.hasNext()) {

                                AbnormalProtocolPoint point = (AbnormalProtocolPoint) points.next();
                                if (abnormalPoints < 15) {
                                    if (point.getRotation() > 700 && point.getSpeed() > 30 && point.getInstrumentSpeed() == 0) {
                                        abnormalPoints++;
                                    } else {
                                        abnormalPoints = 0;
                                    }

                                } else {
                                    return true;
                                }
                            }
                            return false;
                        }
                    });

        return filterTripidsRDD.mapToPair(item -> new Tuple2<>(item._1._1,item._1._2));
    }


    /**
     * 配置命令行参数
     */
    static Options setCmdOptions() {
        Options options = new Options();
        Option opSt = new Option("st", "startTime", true, "[必填参数]开始日期，格式yyyy-MM-dd（包含）");
        Option opEt = new Option("et", "endTime", true, "结束日期，格式yyyy-MM-dd（不包含）");
        opEt.setRequired(false);
        options.addOption(opSt);
        options.addOption(opEt);
        //options.addOption(new Option("tids", "terminalIds", true, "按逗号分隔的terminalId进行回放，如果未设此参数，则为全量数据回放"));
        //options.addOption(new Option("tidsFile", "terminalIdsFile", true, "terminalIds的文件路径，一行一个terminalId"));
        options.addOption(new Option("h", "help", false, "打印帮助"));

        OptionBuilder.withArgName("property=value");
        OptionBuilder.hasArgs(2);
        OptionBuilder.withValueSeparator();
        OptionBuilder.withDescription("其它的配置项，可覆盖配置文件中的配置项(property=value)");
        Option property = OptionBuilder.create("D");
        options.addOption(property);
        return options;
    }

    /**
     * 解析CMD参数，args中通过"-K V"配置属性
     */
    static CommandLine parseCmdOptions(Options options, String[] args) throws ParseException {
        CommandLine cmd = new PosixParser().parse(options, args);
        if (cmd.hasOption("h")) {
            printUsage(options);
            System.exit(0);
        } else if (!cmd.hasOption("st")) {
            printUsage(options);
            System.exit(1);
        }

        return cmd;
    }

    /**
     * 打印帮助信息
     */
    static void printUsage(Options options) {
        HelpFormatter formatter = new HelpFormatter();
        formatter.printHelp("Application", options);
        System.out.println();
    }

    /**
     * 加载config.yaml的配置<br/>
     * CMD中的-D配置能覆盖配置文件中的配置
     */
    static Properties loadProps(CommandLine cmd) throws URISyntaxException, FileNotFoundException {
        //jar包的位置
        File directory = new File(TripAnalysisApplication.class.getProtectionDomain().getCodeSource().getLocation().toURI().getPath()).getParentFile();
        ConfigLoadService configService = new ConfigLoadServiceYaml().setInputStream(new FileInputStream(directory.getAbsolutePath() + "/config.yaml"));
        Properties config = configService.load();

        //命令行中的-D覆盖config.yaml的配置
        Properties cmdProps = cmd.getOptionProperties("D");
        if (cmdProps != null) {
            cmdProps.forEach((k, v) -> config.put(k, v));
        }

        logger.error("OOO:props开始:-----------");
        for (String s : config.stringPropertyNames()) {
            System.out.println(s + "->" + config.getProperty(s));
        }
        logger.error("OOO:props结束：--------------");
        return config;
    }

    /**
     * 解析开始和结束时间<br/>
     * 时间格式：YYYY_MM_DD或YYYYMMDD
     * @param args
     * @param cmd
     */
    static void parseDays(String[] args, CommandLine cmd) {
        if (args.length >= 1) {
            start = DateUtils.parseDate(cmd.getOptionValue("st")).getTime();
            String endTimeStr = cmd.getOptionValue("et");
            end = StringUtils.isEmpty(endTimeStr) ? (start + DAY_MS) :DateUtils.parseDate(endTimeStr).getTime();
        } else {
            throw new RuntimeException("时间不能为空！");
        }
    }

}
